package com.shujia.flink.window;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

public class Demo1TImeWindow {
    public static void main(String[] args) throws Exception {

        /**
         * 时间窗口：由时间触发的窗口
         * SlidingEventTimeWindows: 滑动的事件时间窗口
         * SlidingProcessingTimeWindows：滑动的处理时间窗口
         * TumblingEventTimeWindows：滚动的事件时间窗口
         * TumblingProcessingTimeWindows：滚动的处理时间窗口
         */

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStream<String> linesDS = env.socketTextStream("master", 8888);

        //解析数据，取出时间字段
        DataStream<Tuple2<String, Long>> wordAndTsDS = linesDS.map(line -> {
            String[] split = line.split(",");
            String word = split[0];
            //将时间戳转换成long类型
            long ts = Long.parseLong(split[1]);
            return Tuple2.of(word, ts);
        }, Types.TUPLE(Types.STRING, Types.LONG));

        //设置时间字段和水位线
        DataStream<Tuple2<String, Long>> assDS = wordAndTsDS.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        //水位线前移1秒
                        .<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                        .withTimestampAssigner((kv, ts) -> kv.f1)
        );

        /**
         * 每隔5秒统计最近10秒单词的数量
         */
        SingleOutputStreamOperator<Tuple2<String, Integer>> kvDS = assDS
                .map(kv -> Tuple2.of(kv.f0, 1), Types.TUPLE(Types.STRING, Types.INT));

        kvDS.keyBy(kv -> kv.f0)
                //滑动的事件时间窗口
                .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .sum(1)
                .print();

        env.execute();


    }
}
